/**
 * 
 */
package org.pump.client.pool;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.pump.client.codec.BinlogEvent;
import org.pump.client.codec.EventData;
import org.pump.client.codec.EventHead;
import org.pump.client.codec.QueryEvent;
import org.pump.client.codec.RotateEvent;
import org.pump.client.codec.XidEvent;
import org.pump.client.filter.FilterTube;
import org.pump.client.main.MyProperty;
import org.pump.client.util.MyConstant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author liudong03
 *
 */
public class PumpPool {
	private static final Logger mylog = LoggerFactory
			.getLogger(PumpPool.class);
	LinkedBlockingQueue<EventData> eventQueue;
	long inpos;
	long outpos;
	String out_binlog_file;
	String in_binlog_file;
	MyProperty conf;
	FilterTube filter;
	/**
	 * 
	 */
	public PumpPool(int capacity, MyProperty conf) {
		// TODO Auto-generated constructor stub
		this.eventQueue = new LinkedBlockingQueue<EventData>(capacity);
		
		this.in_binlog_file = conf.binlogFile;
		this.out_binlog_file = conf.binlogFile;
		this.inpos = conf.binlogPosition;
		this.outpos = conf.binlogPosition;
		this.conf = conf;
	}
	
	public void put(BinlogEvent myevent)
	{
		EventHead myhead = myevent.getMyhead();
		EventData mydata = myevent.getMydata();
		
		if(this.filter != null)
		{
			if(!this.filter.doFilter(myhead, mydata))
			{
				mylog.debug("filter event = [" + myevent + "]");
				return;
			}
		}
		
		if(mydata instanceof RotateEvent)
		{
			RotateEvent temp = (RotateEvent)mydata;
			this.in_binlog_file = temp.getNext_binlog_file_name().getStrContent();
			this.inpos = temp.getNext_binlog_event_pos();
		}else
		{
			this.inpos = mydata.getNext_position();
		}
		
		while (true) {
			try {
				if (this.eventQueue.offer(mydata, 1000,
						TimeUnit.MILLISECONDS)) {
					break;
				}
				mylog.error("errno =[" + MyConstant.EVENT_QUEUE_FULL
						+ "]" + " event queue is full");

			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				mylog.info("waitting offer event queue, interrupted exception",e);
			}
		}
	}
	
	public EventData take(long tm)
	{
		EventData queuehead = null;
		while(queuehead == null)
		{
			try {
				queuehead = this.eventQueue.poll(tm, TimeUnit.MILLISECONDS);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				mylog.info("waitting poll event queue, interrupted exception");
			}
			
			if(queuehead instanceof RotateEvent)
			{
				RotateEvent temp = (RotateEvent)queuehead;
				this.out_binlog_file = temp.getNext_binlog_file_name().getStrContent();
				this.outpos = temp.getNext_binlog_event_pos();
				queuehead = null;
			}else if(queuehead instanceof XidEvent)
			{
				this.outpos = queuehead.getNext_position();
				//queuehead = null;
			}else if(queuehead instanceof QueryEvent)
			{
				this.outpos = queuehead.getNext_position();
			}
		}
		
		return queuehead;
	}
	
	public boolean commit()
	{
		this.conf.conf.setProperty("mysql.binlog_position", "" + this.outpos);
		this.conf.conf.setProperty("mysql.binlog_file", this.out_binlog_file);
		if(!this.conf.save())
		{
			mylog.error("commit fail");
			return false;
		}
		return true;
	}
	
	public void setFilterTube(FilterTube mytube)
	{
		this.filter = mytube;
	}
	
}
